package com.ccxe.parser;

import java.io.IOException;
import java.util.Iterator;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MD5Hash;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Partitioner;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;

import com.ccxe.io.Data;
import com.ccxe.io.InjectorKey;
import com.ccxe.util.DATA;
import com.ccxe.util.DateConvertUtil;
import com.ccxe.util.ParserData;
/**
 * 填充分钟数据
 * @author hetao
 *
 */
public class InjectMin {
	public static final Log LOG = LogFactory.getLog(InjectMin.class);

	public static int insert(Path selectPath, Path insertPath)
			throws IOException {
		return run(selectPath, insertPath);
	}

	public static void main(String[] args) throws IOException {
		if (args.length != 2) {
			System.err.println("Usage: com.ccxe.driver.InjectMin <in> <out>");
			System.exit(2);
		}
		System.exit(insert(new Path(args[0]), new Path(args[1])));
	}

	private static int run(Path input, Path output) throws IOException {
		// 如果输出路径存在就删除；
		JobConf conf = new JobConf(InjectMin.class);

		FileInputFormat.addInputPath(conf, input);
		FileOutputFormat.setOutputPath(conf, output);
		
		conf.setJobName("InjectMin_" + DateConvertUtil.getDate());
		conf.setMapperClass(InjectMinMapper.class);
		conf.setMapOutputKeyClass(InjectorKey.class);
		conf.setMapOutputValueClass(Data.class);
		
		conf.setPartitionerClass(InjectMinPartitioner.class);
		conf.setOutputKeyComparatorClass(InjectMinComparator.class);
		conf.setOutputValueGroupingComparator(InjectMinGroupComparator.class);
		
		conf.setReducerClass(InjectMinReducer.class);
		conf.setOutputKeyClass(Data.class);
		conf.setOutputValueClass(NullWritable.class);
		JobClient.runJob(conf);
		return 0;
	}


	public static class InjectMinMapper extends MapReduceBase implements
			Mapper<LongWritable, Text, InjectorKey, Data> {
		
		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<InjectorKey, Data> output, Reporter reporter)
				throws IOException {
			String line = value.toString();
			String[] arrayTmp = ParserData.parser(line);
			// 剔除标题和异常数据
			if (line.indexOf("c_") != -1 || arrayTmp.length < 3)
				return;
			Data data = new Data(arrayTmp);
			// 过滤非交易时间的数据
			int time = (int) ((data.getC_date_time() / 100) % 10000);
			if (key == null
					|| (time >= DATA.OPENINGI_TIME1 && time <= DATA.END_TIME1)
					|| (time >= DATA.OPENINGI_TIME2 && time <= DATA.END_TIME2)) {
				output.collect(new InjectorKey(data.getC_stock_no(),data.getC_date_time()), data);
			}
		}
	}
	
	public static class InjectMinComparator extends WritableComparator {
		protected InjectMinComparator() {
			super(InjectorKey.class, true);
		}
		@Override
		public int compare(Object a, Object b) {
			InjectorKey gk1 = (InjectorKey) a;
			InjectorKey gk2 = (InjectorKey) b;
			return gk1.compareTo(gk2);
		}
	}

	public static class InjectMinGroupComparator extends WritableComparator {
		protected InjectMinGroupComparator() {
			super(InjectorKey.class, true);
		}
		@Override
		public int compare(Object a, Object b) {
			InjectorKey gk1 = (InjectorKey) a;
			InjectorKey gk2 = (InjectorKey) b;
			long l1 = Long.parseLong(gk1.getC_stock_no() + ""
					+ gk1.getC_date_time() / 10000);
			long l2 = Long.parseLong(gk2.getC_stock_no() + ""
					+ gk2.getC_date_time() / 10000);
			return (l1 == l2) ? 0 : (l1 < l2 ? -1 : 1);
		}
	}

	public static class InjectMinPartitioner implements
			Partitioner<InjectorKey, Data> {
		@Override
		public int getPartition(InjectorKey key, Data value, int numPartitions) {
			return Math.abs(MD5Hash.digest(
					key.getC_stock_no() + "" + key.getC_date_time() / 10000)
					.quarterDigest())
					% numPartitions;
		}
		@Override
		public void configure(JobConf jobConfw) {
		}
	}

	public static class InjectMinReducer extends MapReduceBase implements
			Reducer<InjectorKey, Data, Data, NullWritable> {

		private Log LOG = LogFactory.getLog(InjectMinReducer.class);
		private Data tmpData;
		private int count = 0;
		@Override
		public void reduce(InjectorKey key, Iterator<Data> values,
				OutputCollector<Data, NullWritable> output, Reporter reporter)
				throws IOException {
			LOG.info("---------------InjectMinReducer run--------------------");
			while (values.hasNext()) {
				count ++;
				Data data = (Data) values.next();
				if (tmpData == null) {
					tmpData = resetTop(data);
				}
				if (data.getC_date_time() / 100 % 10000 >= DATA.END_TIME1) {
					tmpData = null;
				}
				generData(tmpData, data, output, key);
				if(isValid(data.getC_date_time())) {
					output.collect(data, NullWritable.get());
				}
//				LOG.info(key + "-----------" + data);
				try {
					tmpData = (Data) data.clone();
				} catch (CloneNotSupportedException e) {
					e.printStackTrace();
				}
			}
			if (tmpData.getC_date_time() % 10000 / 100 != 0) {
				Data lastData = resetLast(tmpData);
				generData(tmpData, lastData, output, key);
				if (isValid(lastData.getC_date_time())) {
					output.collect(lastData,NullWritable.get());
				}
			}
			LOG.info("--------------InjectMinReducer hashNext() end-----------------------------");
			LOG.info(" [ "+tmpData.getC_stock_no()+" ]号股票在"+tmpData.getC_date_time()/10000+"内，数据文件中有 【  "+count+"  】条数据");
			tmpData = null;
			count = 0;
			
		}

		// 仅仅支持降序,比较值为分钟，缺省的Data秒全都设置为59
		public void generData(Data data1, Data data2,
				OutputCollector<Data, NullWritable> output, InjectorKey key)
				throws IOException {

			if (data1 == null || data2 == null) {
				return;
			}
			long l1 = (data1.getC_date_time() / 100 * 100 + 59) - 100;
			long l2 = data2.getC_date_time() / 100 * 100 + 59;
			int sno = data1.getC_stock_no();

			for (long i = l1; i > l2; i -= 100) {
				Data tmp = new Data(sno, i);
//				String str = sno + "" + i;
				if (isValid(tmp.getC_date_time())) {
					output.collect(tmp, NullWritable.get());
				}
//				LOG.info(str + "-----User-defined----" + tmp.getC_date_time()
//						% 1000000 / 100);
			}
		}

		public void setTmpData(Data tmpData) {
			this.tmpData = tmpData;
		}

		public Data getTmpData() {
			return tmpData;
		}

		public Data resetTop(Data data) {
			return new Data(data.getC_stock_no(),
					(data.getC_date_time() / 10000) * 10000 + 6059);
		}

		public Data resetLast(Data data) {
			return new Data(tmpData.getC_stock_no(),
					((tmpData.getC_date_time() / 10000) * 10000) + 59);
		}

		private boolean isValid(long time) {
			time = time % 1000000 / 100;
			return (time >= DATA.OPENINGI_TIME1 && time <= DATA.END_TIME1)
					|| (time >= DATA.OPENINGI_TIME2 && time <= DATA.END_TIME2);
		}

	}

}
